package com.atguigu.utils;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/*
从kafka指定主题消费数据，写入kafka指定主题
 */
public class KafkaUtil {

    //从kafka指定主题消费数据
    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic,String groupId){//String：反序列化后的类型
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, GmallConfig.BOOTSTRAP_SERVER);
        //flume的header是写到value里

        return new FlinkKafkaConsumer<>(topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                if (record == null || record.value() == null) {
                    return null;
                }
                return new String(record.value());
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return BasicTypeInfo.STRING_TYPE_INFO;
            }
        }, properties);

    }

    //写数据到kafka指定主题
    public static  FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic){//指定以字符串的格式写入kafka
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,GmallConfig.BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<String>(topic, new SimpleStringSchema(), properties);

    }

    //将未加工的事实表数据动态写到kafka不同主题
    public static FlinkKafkaProducer<JSONObject> getFlinkKafkaProducer(){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,GmallConfig.BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<JSONObject>("", new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject value, @Nullable Long timestamp) {


                 return new ProducerRecord<byte[], byte[]>(value.getString("sink_table"),value.getString("data").getBytes());
                 //将data数据写到sink_table对应的value值的kafka主题里


            }
        }, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    //对上面优化，任何类型的数据都可以写到kafka里
    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,GmallConfig.BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<T>("",kafkaSerializationSchema,properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

    }



    //连接kafka主题创建动态表
    public static String getKafkaTopicDbDDL(String groupId){
        String sql=
         "create table topic_db(" +
                "`database` string," +
                "`table` string," +
                "`type` string," +
                "`data` map<string,string>," +
                "`old` map<string,string>," +
                "`ts` string," +
                "`pt` AS PROCTIME()," +
                "`rt` as TO_TIMESTAMP_LTZ(cast(ts as bigint)*1000,3)," +//interval join要求两表都要指定事件时间和watermark
                "watermark for rt as rt)"+getKafkaConnectorDDL("topic_db",groupId);
        System.out.println(sql);
        return sql;


    }

    public static String getKafkaConnectorDDL(String topic,String groupId){
        return "with(" +
                "'connector'='kafka'," +
                "'topic'='"+topic+"'," +
                "'properties.bootstrap.servers' = '"+GmallConfig.BOOTSTRAP_SERVER+"'," +
                "'properties.group.id' = '"+groupId+"'," +
                "'scan.startup.mode' = 'group-offsets'," +
                "'format'='json')";

    }

    public static String getUpsertKafkaDDL(String topic){
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"+topic+"',\n" +
                "  'properties.bootstrap.servers' = '"+GmallConfig.BOOTSTRAP_SERVER+"',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }


}
